package com.tiantian.kafka.learn_07_idempotence;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.Future;

import static com.tiantian.kafka.Constants.IDEMPOTENCE_TOPIC;

/**
 * @ClassName KafkaIdempotentProducer
 * @Description 幂等性 idempotence ['aɪdəmpoʊtəns],
 * 在创建Kafka生产者时:::设置enable.idempotence为true，并确保acks设置为all（或至少为1）。
 * @Author tant
 * @Date 2025/3/5 13:47
 */
@SuppressWarnings("all")
public class KafkaIdempotentProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.217.132:9092");
        //配置key和value的序列化方式
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 开启幂等性
        props.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, String.valueOf(true));
        // 确保acks设置为all
        props.setProperty(ProducerConfig.ACKS_CONFIG, "all");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        ProducerRecord<String, String> record = new ProducerRecord<>(IDEMPOTENCE_TOPIC, "key", "value");
        //发送消息
        Future<RecordMetadata> future = producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception e) {
                if (e == null) {
                    //发送成功
                    System.out.printf("消息发送成功: topic: %s partition: %d offset: %d key: %s value: %s%n", metadata.topic(), metadata.partition(), metadata.offset(), record.key(), record.value());
                } else {
                    System.out.println("消息发送失败"+e.getMessage());
                }
            }
        });

        producer.close();
    }
}

    